package com.winit.schedule.job.tracker.event.persist;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.event.EventHandler;
import com.winit.schedule.core.repository.constants.TaskItemStatus;
import com.winit.schedule.core.repository.entity.TaskItemEntity;
import com.winit.schedule.core.repository.entity.TaskRunRecordEntity;
import com.winit.schedule.core.utils.Clock;
import com.winit.schedule.job.tracker.JobTrackerContext;
import com.winit.schedule.job.tracker.repository.ScheduleDataManager;

public class TaskItemHandler implements EventHandler<DataPersistEvent> {

    private static final Logger           logger            = LoggerFactory.getLogger(TaskTrackerHandler.class);

    private ScheduleDataManager           scheduleDataManager;

    private JobTrackerContext             jobTrackerContext;

    private List<Thread>                  dataHandlerThread = new ArrayList<Thread>();

    private BlockingQueue<TaskItemEntity> dataQueue         = new LinkedBlockingQueue<TaskItemEntity>();

    private BlockingQueue<TaskItemEntity> recoveryDataQueue = new LinkedBlockingQueue<TaskItemEntity>();

    public TaskItemHandler(JobTrackerContext jobTrackerContext){
        this.scheduleDataManager = jobTrackerContext.getScheduleDataManager();
        this.jobTrackerContext = jobTrackerContext;

        for (int i = 1; i <= 3; i++) {
            Thread thread = new DataHandler(i);
            dataHandlerThread.add(thread);
            thread.start();
        }

        // 必须单线程执行
        RecoveryHandler recoveryHandler = new RecoveryHandler(0);
        dataHandlerThread.add(recoveryHandler);
        recoveryHandler.start();
    }

    @Override
    public void handle(DataPersistEvent event) {
        if (DataPersistEventType.TASKITEM_STATUS_UPDATE.equals(event.getType()) && null != event.getTaskItemEntities()) {
            for (TaskItemEntity taskItemEntity : event.getTaskItemEntities()) {
                boolean isSuccess = dataQueue.offer(taskItemEntity);
                if (!isSuccess) {
                    logger.warn(">>>>>>>>>> Record run log fails, the queue is full <<<<<<<<<<");
                }
            }
        } else if (DataPersistEventType.RECOVERY_TASKITEM.equals(event.getType())
                   && null != event.getTaskItemEntities()) {
            for (TaskItemEntity taskItemEntity : event.getTaskItemEntities()) {
                boolean isSuccess = recoveryDataQueue.offer(taskItemEntity);
                if (!isSuccess) {
                    logger.warn(">>>>>>>>>> Record recovery taskitem fails, the queue is full <<<<<<<<<<");
                }
            }
        }
    }

    public ScheduleDataManager getScheduleDataManager() {
        return scheduleDataManager;
    }

    public JobTrackerContext getJobTrackerContext() {
        return jobTrackerContext;
    }

    /**
     * 写入任务运行日志，可多线程执行
     * 
     * @version 1.2.0
     * <pre>
     * Author	Version		Date		Changes
     * jianke.zhang 	1.0  		2015年8月4日 	Created
     *
     * </pre>
     * @since 1.
     */
    class DataHandler extends Thread {

        public DataHandler(int num){
            this.setName("Thread-TaskItem-Persist-" + num);
        }

        @SuppressWarnings({ "unchecked", "static-access" })
        @Override
        public void run() {
            while (true) {
                try {
                    TaskItemEntity taskItem = dataQueue.take();
                    if (null != taskItem) {
                        if (TaskItemStatus.ACTIVIE.equals(taskItem.getTaskItemStatus())) {
                            scheduleDataManager.updateTaskRunnerForTaskItem(taskItem);

                            // 取消记录运行日志
                            /*TaskRunRecordEntity taskRunRecord = new TaskRunRecordEntity(taskItem.getTaskRunner(),
                                taskItem.getTaskItemStatus());
                            jobTrackerContext.getDataDispatcher()
                                .getEventHandler()
                                .handle(new DataPersistEvent(DataPersistEventType.LOG_RUN_RECORED, taskRunRecord));*/
                        } else if (TaskItemStatus.COMPLETE.equals(taskItem.getTaskItemStatus())
                                   || TaskItemStatus.STOP.equals(taskItem.getTaskItemStatus())) {

                            // 取消记录运行日志
                            /*TaskRunRecordEntity taskRunRecord = new TaskRunRecordEntity(taskItem.getTaskRunner(),
                                Clock.getSystemTime(),
                                taskItem.getTaskItemStatus());
                            jobTrackerContext.getDataDispatcher()
                                .getEventHandler()
                                .handle(new DataPersistEvent(DataPersistEventType.LOG_RUN_RECORED, taskRunRecord));*/
                        }
                    }

                    Thread.currentThread().sleep(50);
                } catch (Exception e) {
                    logger.error("TaskItem persist failure.", e);
                }
            }
        }
    }

    /**
     * 恢复任务项,单线程任务
     * 
     * @version 1.2.0
     * <pre>
     * Author	Version		Date		Changes
     * jianke.zhang 	1.0  		2015年8月4日 	Created
     *
     * </pre>
     * @since 1.
     */
    class RecoveryHandler extends Thread {

        public RecoveryHandler(int num){
            this.setName("Thread-TaskItem-recoveryHandler-" + num);
        }

        @SuppressWarnings("static-access")
        @Override
        public void run() {
            while (true) {
                try {
                    TaskItemEntity taskItem = recoveryDataQueue.take();
                    if (null != taskItem) {
                        if (TaskItemStatus.STOP.equals(taskItem.getTaskItemStatus())) {
                            scheduleDataManager.updateTaskItemStatusByTaskRunnerUuid(taskItem);
                        }
                    }

                    Thread.currentThread().sleep(50);
                } catch (Exception e) {
                    logger.error("TaskItem persist failure.", e);
                }
            }
        }
    }

}
